"""
消息队列与并发控制
负责管理弹幕消息队列和处理流程
"""

import queue
import threading
import time
from typing import Callable, Optional, Dict

from ..core.models import DanmakuMessage, SystemStatus
from ..utils.logger import get_logger


class MessageQueue:
    """消息队列管理器"""

    def __init__(
        self,
        max_size: int = 100,
        max_messages_per_second: int = 3,
        enable_priority: bool = True
    ):
        """
        初始化消息队列
        
        Args:
            max_size: 队列最大容量
            max_messages_per_second: 每秒最多处理消息数
            enable_priority: 是否启用优先级队列
        """
        self.max_size = max_size
        self.max_messages_per_second = max_messages_per_second
        self.enable_priority = enable_priority

        # 使用优先级队列
        if enable_priority:
            self.queue = queue.PriorityQueue(maxsize=max_size)
        else:
            self.queue = queue.Queue(maxsize=max_size)

        self.logger = get_logger()

        # 统计信息
        self.total_enqueued = 0
        self.total_dequeued = 0
        self.total_dropped = 0

    def enqueue(self, message: DanmakuMessage) -> bool:
        """
        将消息加入队列
        
        Args:
            message: 弹幕消息
            
        Returns:
            是否成功加入队列
        """
        try:
            if self.enable_priority:
                # 优先级队列:数字越小优先级越高,所以取负值
                priority = -message.priority
                self.queue.put_nowait((priority, self.total_enqueued, message))
            else:
                self.queue.put_nowait(message)

            self.total_enqueued += 1
            self.logger.debug(f"消息入队 [{message.username}]: {message.content[:30]}")
            return True

        except queue.Full:
            self.total_dropped += 1
            self.logger.warning(f"消息队列已满,丢弃消息: [{message.username}] {message.content[:30]}")
            return False

    def dequeue(self, timeout: Optional[float] = None) -> Optional[DanmakuMessage]:
        """
        从队列取出消息
        
        Args:
            timeout: 超时时间(秒),None表示阻塞等待
            
        Returns:
            弹幕消息,如果队列为空且超时则返回None
        """
        try:
            if self.enable_priority:
                _, _, message = self.queue.get(timeout=timeout)
            else:
                message = self.queue.get(timeout=timeout)

            self.total_dequeued += 1
            return message

        except queue.Empty:
            return None

    def size(self) -> int:
        """获取当前队列长度"""
        return self.queue.qsize()

    def is_empty(self) -> bool:
        """队列是否为空"""
        return self.queue.empty()

    def is_full(self) -> bool:
        """队列是否已满"""
        return self.queue.full()

    def clear(self):
        """清空队列"""
        while not self.queue.empty():
            try:
                self.queue.get_nowait()
            except queue.Empty:
                break
        self.logger.info("消息队列已清空")

    def get_stats(self) -> Dict[str, int]:
        """获取统计信息"""
        return {
            "size": self.size(),
            "total_enqueued": self.total_enqueued,
            "total_dequeued": self.total_dequeued,
            "total_dropped": self.total_dropped
        }


class MessageProcessor:
    """消息处理器(生产者-消费者模式)"""

    def __init__(
        self,
        message_queue: MessageQueue,
        process_callback: Callable[[DanmakuMessage], None],
        min_process_interval: float = 0.3,  # 最小处理间隔(秒)
        user_reply_interval: float = 3.0     # 同一用户最小回复间隔(秒)
    ):
        """
        初始化消息处理器
        
        Args:
            message_queue: 消息队列
            process_callback: 消息处理回调函数
            min_process_interval: 最小处理间隔(控制处理速度)
            user_reply_interval: 同一用户最小回复间隔
        """
        self.message_queue = message_queue
        self.process_callback = process_callback
        self.min_process_interval = min_process_interval
        self.user_reply_interval = user_reply_interval

        self.logger = get_logger()

        # 处理线程
        self.process_thread: Optional[threading.Thread] = None
        self.is_running = False

        # 用户回复时间记录
        self.user_last_reply_time: Dict[str, float] = {}

        # 最后处理时间
        self.last_process_time = 0.0

    def start(self):
        """启动消息处理线程"""
        if self.is_running:
            self.logger.warning("消息处理器已在运行")
            return

        self.is_running = True
        self.process_thread = threading.Thread(target=self._process_loop, daemon=True)
        self.process_thread.start()
        self.logger.info("✓ 消息处理器已启动")

    def stop(self):
        """停止消息处理"""
        self.is_running = False
        if self.process_thread:
            self.process_thread.join(timeout=5)
        self.logger.info("✓ 消息处理器已停止")

    def _process_loop(self):
        """消息处理循环"""
        while self.is_running:
            try:
                # 从队列取出消息(阻塞1秒)
                message = self.message_queue.dequeue(timeout=1.0)

                if message is None:
                    continue

                # 检查处理间隔
                current_time = time.time()
                time_since_last_process = current_time - self.last_process_time

                if time_since_last_process < self.min_process_interval:
                    # 等待到最小间隔
                    time.sleep(self.min_process_interval - time_since_last_process)

                # 检查同一用户回复间隔
                if not self._check_user_interval(message.user_id):
                    self.logger.debug(f"用户 {message.username} 回复间隔过短,跳过")
                    continue

                # 处理消息
                self.logger.info(f"正在处理消息 [{message.username}]: {message.content}")
                self.process_callback(message)

                # 更新处理时间
                self.last_process_time = time.time()
                self.user_last_reply_time[message.user_id] = time.time()

            except Exception as e:
                self.logger.error(f"处理消息异常: {e}")

    def _check_user_interval(self, user_id: str) -> bool:
        """
        检查用户回复间隔
        
        Args:
            user_id: 用户ID
            
        Returns:
            是否满足间隔要求
        """
        if user_id not in self.user_last_reply_time:
            return True

        last_time = self.user_last_reply_time[user_id]
        elapsed = time.time() - last_time

        return elapsed >= self.user_reply_interval


class ConcurrentController:
    """并发控制器"""

    def __init__(self):
        """初始化并发控制器"""
        self.logger = get_logger()

        # 线程池
        self.threads: Dict[str, threading.Thread] = {}

        # 系统状态
        self.status = SystemStatus()

    def start_thread(
        self,
        name: str,
        target: Callable,
        daemon: bool = True
    ) -> bool:
        """
        启动一个线程
        
        Args:
            name: 线程名称
            target: 线程目标函数
            daemon: 是否为守护线程
            
        Returns:
            是否成功启动
        """
        try:
            if name in self.threads and self.threads[name].is_alive():
                self.logger.warning(f"线程 {name} 已在运行")
                return False

            thread = threading.Thread(target=target, name=name, daemon=daemon)
            thread.start()
            self.threads[name] = thread

            self.logger.info(f"✓ 线程 {name} 已启动")
            return True

        except Exception as e:
            self.logger.error(f"启动线程 {name} 失败: {e}")
            return False

    def stop_thread(self, name: str, timeout: float = 5.0):
        """
        停止线程
        
        Args:
            name: 线程名称
            timeout: 等待超时时间(秒)
        """
        if name not in self.threads:
            return

        thread = self.threads[name]
        if thread.is_alive():
            thread.join(timeout=timeout)

            if thread.is_alive():
                self.logger.warning(f"线程 {name} 未在超时时间内停止")
            else:
                self.logger.info(f"✓ 线程 {name} 已停止")

        del self.threads[name]

    def stop_all_threads(self, timeout: float = 5.0):
        """停止所有线程"""
        for name in list(self.threads.keys()):
            self.stop_thread(name, timeout)

    def get_thread_status(self) -> Dict[str, bool]:
        """获取所有线程状态"""
        return {
            name: thread.is_alive()
            for name, thread in self.threads.items()
        }

    def update_status(self, **kwargs):
        """
        更新系统状态
        
        Args:
            **kwargs: 要更新的状态字段
        """
        for key, value in kwargs.items():
            if hasattr(self.status, key):
                setattr(self.status, key, value)

    def get_status(self) -> SystemStatus:
        """获取系统状态"""
        return self.status
